package com.flink.flinkdemo.demo;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.CoGroupOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author zhangy on 2020/12/31
 * @description: add you description
 */
public class CountBlockInfo {


    public static void main(String[] args) throws Exception {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> dataSource = env.readTextFile("/Users/zhangyong/Idea-git/flink-d/flink-demo/src/doc/newapi");
        FlatMapOperator<String, Tuple2<String, String>> operator_account = dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, String>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, String>> out) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {
                    if (s.contains("Operator_Account")) {
                        String[] split1 = s.split(":");
                        if(split1.length>1){
                            out.collect(new Tuple2<>(split1[0].replace("\\\"",""), split1[1].replace("\\\"","")));
                        } else {
                            System.out.println("size < 2"+s);
                        }
                    }
                }
            }
        });
//        operator_account.print();


        DataSource<String> riskLog = env.readTextFile("/Users/zhangyong/Idea-git/flink-d/flink-demo/src/doc/fengkong");
        MapOperator<String, String> map = riskLog.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                String bind = null;
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String message = jsonObject.getString("message");
                    String replace = message.split(" ")[3].replace("【结果】:", "");
                    bind = JSONObject.parseObject(replace).getJSONObject("data").getString("bind");
                } catch (Exception e) {
//                    System.out.println("异常记录："+value);
//                    System.out.println("异常记录---处理后"+value.split(" ")[0]);
                    return "";
                }
                return bind;
            }
        });
        FlatMapOperator<String, Tuple2<String, String>> flatMapRisk = map.flatMap(new FlatMapFunction<String, Tuple2<String, String>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, String>> out) throws Exception {
                if(JSONObject.isValid(value)){
                    String userinfouri = JSONObject.parseObject(value).getString("userinfouri");
                    out.collect(new Tuple2<>(userinfouri,value));
                }

            }
        });

//        flatMapRisk.print();

        operator_account.coGroup(flatMapRisk).where(0).equalTo(0).with(new CoGroupFunction<Tuple2<String, String>, Tuple2<String, String>, Tuple2<String,String>>() {
            @Override
            public void coGroup(Iterable<Tuple2<String, String>> first, Iterable<Tuple2<String, String>> second, Collector<Tuple2<String,String>> out) throws Exception {
                for (Tuple2<String, String> f : first) {
                    for (Tuple2<String, String> s : second) {
                        String tv = "";
                        if(f.f1.equals(s.f0)){
                            tv = s.f1;
                        }
                        out.collect(new Tuple2<>(f.f1,s.f1));
                    }
                }
            }
        }).print();



    }
}